package data.platform.rest;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import data.platform.common.service.command.MergeMessageService;
import data.platform.common.service.query.DataAggregateQueryService;
import data.platform.rest.domain.AggregateQuery;
import data.platform.rest.domain.AggregateResult;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;

@RestController
@RequestMapping("/api/v1/merge")
@Slf4j
@RequiredArgsConstructor
public class MergeController {

    final DataAggregateQueryService dataAggregateQueryService;

    final MergeMessageService mergeMessageService;

    private ObjectMapper objectMapper;

    @PostConstruct
    public void init() {
        objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
        objectMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
    }

    @PostMapping(value = "/aggregate/query")
    public Mono<ResponseEntity<List<AggregateResult>>> aggregateQuery(@RequestBody AggregateQuery aggregateQuery) {
        return dataAggregateQueryService.queryAggregateItems(aggregateQuery.getArithmeticType(), aggregateQuery.getMetric(), aggregateQuery.getTags())
                .map(result -> {
                    Map<String, String> tag = null;
                    Map<String, Object> aggregateItemMap = null;
                    try {
                        tag = objectMapper.readValue(result.getT1(), Map.class);
                        aggregateItemMap = objectMapper.convertValue(result.getT2(), Map.class);
                    } catch (Exception e) {
                        log.error("", e);
                    }
                    return new AggregateResult(tag, aggregateItemMap);
                })
                .collectList()
                .elapsed()
                .map(tuple2 -> ResponseEntity.ok()
                        .header("executetime", tuple2.getT1().toString())
                        .body(tuple2.getT2()));
    }

    @GetMapping(value="mergeMessage")
    public void sendMergeMessage() {
        mergeMessageService.sendMergeMessage("success");
    }
}
